package yuekao1.util;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;
import yuekao1.entity.*;

import java.util.Collections;

public class AsyncIORedisUtil extends RichAsyncFunction<Tuple2<OrderInfo, OrderDetail>,Tuple5<OrderInfo, OrderDetail, BaseProvince, SkuInfo, UserInfo>> {

    Jedis jedis = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = new Jedis("hadoop-single");
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }

    @Override
    public void asyncInvoke(Tuple2<OrderInfo, OrderDetail> value, ResultFuture<Tuple5<OrderInfo, OrderDetail, BaseProvince, SkuInfo, UserInfo>> out) throws Exception {
        String s = jedis.get("base_province:" + value.f0.getProvince_id());
//        System.out.println(s);
        String data = JSON.parseObject(s).getString("data");
        BaseProvince baseProvince = JSON.parseObject(data, BaseProvince.class);

        String s1 = jedis.get("user_info:" + value.f0.getUser_id());
        String data1 = JSON.parseObject(s1).getString("data");
        UserInfo userInfo = JSON.parseObject(data1, UserInfo.class);

        String s2 = jedis.get("sku_info:" + value.f1.getSku_id());
        String data2 = JSON.parseObject(s2).getString("data");
        SkuInfo skuInfo = JSON.parseObject(data2, SkuInfo.class);
        out.complete(Collections.singleton(new Tuple5<>(value.f0, value.f1, baseProvince, skuInfo, userInfo)));

    }
       
}
